package com.ustc.hewei.rpc.client.loadbalance;

import com.ustc.hewei.rpc.client.ClientProxyFactory;
import com.ustc.hewei.rpc.client.discovery.ServiceDiscovery;
import com.ustc.hewei.rpc.common.Utils.MessageSerialize;
import com.ustc.hewei.rpc.common.protocol.NotifyMessage;
import com.ustc.hewei.rpc.common.protocol.NotifyType;
import com.ustc.hewei.rpc.common.protocol.Service;
import com.ustc.hewei.rpc.server.register.notify.NotifyRegister;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

/**
* @description: 服务的主动通知更新
* @author hewei
* @date 2022/8/23 09:04
* @version 1.0
*/

@Slf4j
public class UpdateService implements Runnable{
    private final NotifyRegister notifyRegister;

    // 用以存放待处理的失效服务的阻塞队列
    private final BlockingQueue<Service> invalidService;

    private final ServiceDiscovery serviceDiscovery;

    // 客户端本地缓存列表
    private final ConcurrentHashMap<String, List<Service>> cacheList;

    private final ClientProxyFactory clientProxyFactory;

    public UpdateService(NotifyRegister notifyRegister, ServiceGetter serviceGetter
            , ServiceDiscovery serviceDiscovery, ConcurrentHashMap<String, List<Service>> cacheList,
                         ClientProxyFactory clientProxyFactory) {
        this.notifyRegister = notifyRegister;
        invalidService = serviceGetter.getInvalidService();
        this.serviceDiscovery = serviceDiscovery;
        this.cacheList = cacheList;
        Executors.newFixedThreadPool(2).submit(this);
        this.clientProxyFactory = clientProxyFactory;
    }

    @Override
    /*
     * @description: 从阻塞队列拿取任务进行处理
     * @author: hewei
     * @date: 2022/8/23 22:02
     **/
    public void run() {
        while (true) {
            try {
                Service service = invalidService.take();
                sendNotify(service);
                String serviceName = service.getName();
                List<Service> services = serviceDiscovery.getServices(serviceName);
                cacheList.put(serviceName, services);
                clientProxyFactory.getUnService().remove(service);
                log.info("已从注册中心强制刷新服务列表缓存成功！");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /*
     * @description: 将包含失效服务的通知发送到注册中心
     * @param service
     * @author: hewei
     * @date: 2022/8/23 22:03
     **/
    public void sendNotify(Service service) {
        NotifyMessage notifyMessage = new NotifyMessage();
        String key = "rpc/" + service.getName() + "/service";
        notifyMessage.setService(service).setKey(key).setStatus(NotifyType.UPDATE);
        byte[] data = MessageSerialize.objectToByteArray(notifyMessage);
        byte[] resp = notifyRegister.notifySend(data, service);
        NotifyMessage message = MessageSerialize.byteArrayToNotify(resp);
        if (message.getStatus() == NotifyType.SUCCESS) {
            log.info("服务 {} 通知注册中心更新成功！", service);
        } else log.error("服务 {} 通知注册中心更新失败！", service);
    }
}
